#include "stream_service.h"
#include <cstring>
#include <iostream>
#include <chrono>
#include "event2/event.h"
#include "event2/listener.h"
#include "event2/bufferevent.h"
#include "../com/simple_logger.h"
#include "../security_msg.pb.h"
#include "../com/utils.h"
#include "../com/file_utils.h"
#include "stream_handle.h"
#include "../globe_type.h"

using namespace std;
using namespace sec;

static const char *STORAGE = "temp/";

static void SConnListenerCb(struct evconnlistener *listen, evutil_socket_t sock, struct sockaddr *addr, int len, void *ctx)
{
    // 解析客户端ip
    char ip[16] = {0};
    sockaddr_in *addr_in = (sockaddr_in *)addr;
    evutil_inet_ntop(AF_INET, &addr_in->sin_addr, ip, sizeof(ip));
    stringstream ss;
    ss << ip << ":" << addr_in->sin_port << " 流连接接受...";
    LOG4CPLUS_INFO(SimpleLogger::Get()->LoggerRef(), ss.str());
    StreamService *service = (StreamService *)ctx;
    int s = sock;

    service->ConnListener(s);
}

static void SBuffeventReadCb(struct bufferevent *b, void *ctx)
{
    StreamService *service = (StreamService *)ctx;
    service->ReadCb();
}

static void SBuffereventWriteCb(struct bufferevent *b, void *ctx)
{
    StreamService *service = (StreamService *)ctx;
    service->WriteCb();
}

static void SBuffereventEventCb(struct bufferevent *b, short which, void *ctx)
{
    StreamService *service = (StreamService *)ctx;
    service->EventCb(which);
}

StreamService::StreamService(int port, int heartbeat)
    : port_(port), heartbeat_(heartbeat), quit_flag_(false)
{
}

StreamService::~StreamService()
{
    Stop();
}

bool StreamService::Init()
{
    // 创建事件循环
    ev_base_ = event_base_new();
    if (!ev_base_)
    {
        return false;
    }
    sockaddr_in sin;
    memset(&sin, 0, sizeof(sin));
    sin.sin_family = AF_INET;
    sin.sin_port = htons(port_);
    // 创建socket连接回调
    ev_listener_ = evconnlistener_new_bind(
        ev_base_,
        SConnListenerCb,
        this,
        LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE,
        16,
        (sockaddr *)&sin,
        sizeof(sin));
    if (!ev_listener_)
    {
        return false;
    }
    FileUtils futil;
    if (!futil.Exist(STORAGE))
    {
        futil.MDir(STORAGE);
    }
    return true;
}

bool StreamService::Connect(const char *ip, int port)
{
    buf_ev_ = bufferevent_socket_new(ev_base_, -1, BEV_OPT_CLOSE_ON_FREE);
    if (!buf_ev_)
    {
        return false;
    }
    bufferevent_setcb(buf_ev_,
                      SBuffeventReadCb,
                      SBuffereventWriteCb,
                      SBuffereventEventCb,
                      this);
    bufferevent_enable(buf_ev_, EV_READ | EV_WRITE);
    timeval tv = {this->heartbeat_, 0};
    bufferevent_set_timeouts(buf_ev_, &tv, &tv);

    sockaddr_in sin;
    memset(&sin, 0, sizeof(sin));
    sin.sin_family = AF_INET;
    sin.sin_port = htons(port);
    evutil_inet_pton(AF_INET, ip, &sin.sin_addr.s_addr);
    int ret = bufferevent_socket_connect(buf_ev_, (sockaddr *)&sin, sizeof(sin));
    if (ret != 0)
    {
        return false;
    }
    return true;
}

void StreamService::ReadCb()
{
    char flag[5] = {0};
    int head_size = 0;
    // 判断报文头
    int len = bufferevent_read(buf_ev_, flag, 4);
    if (len <= 0 || strcmp(flag, PKT_FLAG) != 0)
    {
        return;
    }
    // 获得消息头大小
    len = bufferevent_read(buf_ev_, &head_size, 4);
    if (len <= 0)
    {
        return;
    }
    char *head = new char[head_size];
    len = bufferevent_read(buf_ev_, head, head_size);
    // 解析消息头对象
    MsgHeader msg_header;
    msg_header.ParseFromArray(head, head_size);
    delete[] head;
    char *msg_buff = new char[msg_header.msg_size()];
    // FIX:if msg_body too large
    len = bufferevent_read(buf_ev_, msg_buff, msg_header.msg_size());
    switch (msg_header.msg_type())
    {
    case MsgType::CONNECT_RESP: // ignore
        break;
    default:
        LOG4CPLUS_ERROR(SimpleLogger::Get()->LoggerRef(), " 消息类型处理未定义");
        SendHeartBeat();
    }
    delete[] msg_buff;
}

void StreamService::WriteCb()
{
}

void StreamService::EventCb(short which)
{
    if (which & BEV_EVENT_CONNECTED)
    {
        // 当与主服务连接完成以后发送注册报文
        LOG4CPLUS_DEBUG(SimpleLogger::Get()->LoggerRef(), "StreamService与主服务连接完成");
        ServiceRegist regist;
        regist.set_service_name("stream_service");

        MsgHeader header;
        header.set_from(0);
        header.set_to(0);
        header.set_msg_type(MsgType::SERVICE_REGIST);
        header.set_msg_size(regist.ByteSizeLong());

        SendToCenter(header.SerializeAsString().c_str(), header.ByteSizeLong(), regist.SerializeAsString().c_str(), regist.ByteSizeLong());
    }
    if (which & BEV_EVENT_TIMEOUT) // timeout
    {
        SendHeartBeat();
        bufferevent_enable(buf_ev_, EV_READ | EV_WRITE);
    }
    if (which & BEV_EVENT_EOF)
    {
        LOG4CPLUS_DEBUG(SimpleLogger::Get()->LoggerRef(), "StreamService 连接中断");
    }
}

void StreamService::Start()
{
    if (!run_thread_)
    {
        quit_flag_ = false;
        // 在线程中分发
        run_thread_ = new thread(&StreamService::Dispatch, this);
    }
}

void StreamService::Stop()
{
    if (run_thread_)
    {
        // 退出线程和事件循环
        quit_flag_ = true;
        run_thread_->join();
        delete run_thread_;
        run_thread_ = nullptr;
    }
}

void StreamService::Dispatch()
{
    stringstream ss;
    ss << "StreamService 监听端口：" << port_;
    LOG4CPLUS_INFO(SimpleLogger::Get()->LoggerRef(), ss.str());
    // 事件分发并循环处理
    while (!quit_flag_)
    {
        event_base_loop(ev_base_, EVLOOP_NONBLOCK);
        this_thread::sleep_for(chrono::milliseconds(1));
    }
    bufferevent_free(buf_ev_);
    evconnlistener_free(ev_listener_);
    event_base_free(ev_base_);
    buf_ev_ = nullptr;
    ev_base_ = nullptr;
    ev_listener_ = nullptr;
}

void StreamService::ConnListener(int socket)
{
    StreamHandle *handler = new StreamHandle(socket, this);
    thread t(&StreamHandle::Launch, handler);
    t.detach();
}

bool StreamService::SendToCenter(const char *header, int header_size, const char *data, int data_size)
{
    mtx_.lock();
    int ret = bufferevent_write(buf_ev_, PKT_FLAG, 4);
    if (ret == -1)
    {
        mtx_.unlock();
        return false;
    }
    ret = bufferevent_write(buf_ev_, &header_size, 4);
    if (ret == -1)
    {
        mtx_.unlock();
        return false;
    }
    ret = bufferevent_write(buf_ev_, header, header_size);
    if (ret == -1)
    {
        mtx_.unlock();
        return false;
    }
    if (data_size > 0)
    {
        ret = bufferevent_write(buf_ev_, data, data_size);
        if (ret == -1)
        {
            mtx_.unlock();
            return false;
        }
    }
    mtx_.unlock();
    return true;
}

thread* StreamService::GetThread() const
{
    return run_thread_;
}

void StreamService::SendHeartBeat()
{
    // heartbeat
    chrono::time_point<chrono::system_clock, std::chrono::milliseconds> tp = chrono::time_point_cast<chrono::milliseconds>(chrono::system_clock::now());
    HeartBeat heartBeat;
    heartBeat.set_timestamp(tp.time_since_epoch().count());

    MsgHeader header;
    header.set_from(0);
    header.set_to(0);
    header.set_msg_type(MsgType::HEART_BEAT);
    header.set_msg_size(heartBeat.ByteSizeLong());

    SendToCenter(header.SerializeAsString().c_str(), header.ByteSizeLong(), heartBeat.SerializeAsString().c_str(), heartBeat.ByteSizeLong());
}
